/*
 * Copyright 2007 - The JDPF Project Team (http://www.jdpf.org) 
 * 
 * Licensed under the Apache License, Version 2.0 (the "License"); 
 * you may not use this file except in compliance with the License. 
 * You may obtain a copy of the License at 
 * 
 * http://www.apache.org/licenses/LICENSE-2.0 
 * 
 * Unless required by applicable law or agreed to in writing, software 
 * distributed under the License is distributed on an "AS IS" BASIS, 
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and 
 * limitations under the License. 
 */
package org.jdpf.plugins.file.generator.textseries.delimitedfilereader;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.URI;
import java.net.URISyntaxException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

import org.jdpf.core.kernel.elements.Token;
import org.jdpf.core.plugins.blocks.AbstractGeneratorPlugin;
import org.jdpf.core.plugins.exceptions.PluginException;
import org.jdpf.core.plugins.exceptions.PluginExceptionsEnum;
import org.jdpf.plugins.datatypes.basic.TextAdapter;
import org.jdpf.plugins.datatypes.basic.TextSeries;
import org.jdpf.plugins.datatypes.temporal.TimestampedTextAdapter;

/**
 * This plugin is a generator that reads integer values from a csv file, stores
 * them into a IntegerSeries data type. This series is finally inclosed to a
 * token and added to all output arc places.
 * 
 * @author Paolo Ciccarese
 * @version 0.1
 */

public class DelimitedFileTextSeriesReader extends AbstractGeneratorPlugin {
    
	/*
	 * Parameters list. We prefer to define the plugin parameters with an 
	 * underscore. For every plugin parameter it is necessary to implement 
	 * a 'set' method that is accepting a String variable. The framework,
	 * through reflection, will set the parameters at runtime. 
	 */
    private URI _inputFile; 									// Mandatory 
    
    /**
     * <p>The line format specifies the data to read and their order. 
     * It can include three elements:</p>
     * <ul>
     * 		<li>d - data</li>
     * 		<li>t - timestamp</li>
     * 		<li>u - unit of measure</li>
     * </ul>
     * <p>The default is a file with data and timestamp (without unit 
     * of measure).</p>
     */
    private String _lineFormat = "dt";							// Optional
    private String _timestampFormat = "yyyy-MM-dd HH:mm:ss"; 	// Optional
    private String _tokensSeparator = "\t"; 					// Optional
    
    /**
     * @param fileName	The file to read
     */
	public void setInputFile(URI fileName) {
		_inputFile = fileName;
	}

	/**
	 * If the file separator is defined as 'tab'
	 * we substitute the right character.
	 * @param separator	The file separator
	 */
	public void setTokensSeparator(String separator) {
		if (separator.equalsIgnoreCase("tab"))
			separator = Character.toString('\u0009');
		_tokensSeparator = separator;
	}

	public void setLineFormat(String lineFormat) {
		_lineFormat = lineFormat;
	}
		
	/**
	 * The timestamp format
	 * @param timestamp_format
	 */
	public void setTimestampFormat(String timestampFormat) {
		if (timestampFormat != null) _timestampFormat = timestampFormat;
	}
    
	/**
	 * The data format in between: {int, long, float, double, string}
	 * @param dataFormat	
	 * @throws PluginException
	 */
     
	/*
	 * Standard plugin constructor: every plugin has to implement the 
	 * constructor in this very same format.
	 */
    public DelimitedFileTextSeriesReader(String transitionId, String transitionName) {
		super(transitionId, transitionName);
	}	
    
    @SuppressWarnings("unchecked")
	public Token process(List<Token> list) throws PluginException {
		if(_tokensSeparator.equals(" "))
			throw new PluginException(PluginExceptionsEnum.ExecutionError, this.getPluginId(), "The use of space as tokens separator is not allowed.");

    	TextSeries iSeries = readSeriesFromFile();
    	return setOutputToken(iSeries);
    }
	
    /**
     * @param fileName
     * @return
     * @throws PluginException 
     * @throws URISyntaxException 
     */
    private TextSeries readSeriesFromFile() throws PluginException {
    	
    	// Elements' position detection
    	int dataPosition = _lineFormat.indexOf('d');
    	int timePosition = _lineFormat.indexOf('t');
    	
    	// Timestamp format.
    	SimpleDateFormat sdf = null;
    	if (timePosition!=-1 && _timestampFormat != null) {
    		sdf = new SimpleDateFormat(_timestampFormat);
    	}
    	
    	TextSeries outputSeries = new TextSeries();
    
		try {
			BufferedReader br = null;
			String[] tokens = null;
			
			long fileLength = 0;
			String schema=_inputFile.getScheme();
			if(schema.equalsIgnoreCase("file")){
				FileReader f2 = new FileReader(new File (_inputFile));
				fileLength = (new File(_inputFile)).length();
				br = new BufferedReader(f2);
			}else if(schema.equalsIgnoreCase("http")){
				InputStream stream = _inputFile.toURL().openStream();
				br = new BufferedReader(new InputStreamReader(stream));
			}else
				throw new PluginException(PluginExceptionsEnum.IllegalParameter,this.getPluginId(),
						"Protocol '"+schema+"' not allowed.");			
			
			// File lenght and counter for progress percentage
			this.log(this, this.getPluginId(), this.getId(), "File size: "+fileLength);
			int processedBytes = 0;
			int lineCounter = 1;
			
			this.logMainPoint(this, this.getPluginId(), this.getId(), "Reading  started.");
			
			String readLine = null;
			TextAdapter value = null;
			Date timestamp = null;
			
			while (start && (readLine = br.readLine()) != null) {
				this.log(this, this.getPluginId(), this.getId(), "Line '" + lineCounter + "' in: " + readLine);
				
				// Updating progress percentage
				processedBytes+=readLine.length()+2;
				percentage = Math.min(100, ((double)processedBytes / fileLength) * 100);
				
				tokens = readLine.split(_tokensSeparator); 
				
				// TODO Checks tokens against line format
				if (timePosition == -1) {
					value = getDataValue(tokens[dataPosition], lineCounter);
					outputSeries.addElement(value);
				} else if (timePosition	 != -1) {
					timestamp = getTimestampValue(tokens[timePosition], sdf, lineCounter);
					value = getTimestampedDataValue(tokens[dataPosition], lineCounter, timestamp);
					outputSeries.addElement(value);
				}
				
				this.log(this, this.getPluginId(), this.getId(), "Parsed line '" + lineCounter + "' value in: " + value.getStringValue() );
				
				lineCounter++;
			}
			
			this.logMainPoint(this, this.getPluginId(), this.getId(), "Reading completed!");
			br.close();
		} catch (IOException e) {
			this.error(this, this.getPluginId(), this.getId(), "The input file " + _inputFile + " does not exist: "
					+ e.getMessage());
			throw new PluginException(PluginExceptionsEnum.ExecutionError, this.getPluginId(), "The input file " + _inputFile + " does not exist: "
					+ e.getMessage());
		}

		return outputSeries;
	}
    
    private TextAdapter getDataValue(String value,  int line) throws PluginException {
		try {
			return new TextAdapter(value);
						
		} catch (NumberFormatException e) {
			this.error(this, this.getPluginId(), this.getId(), "Value '" + value + "' in line '" + line + "' is not of type: ");
			throw new PluginException(PluginExceptionsEnum.InputTypeNotRecognized, this.getPluginId(), "Value '" + value + "' in line '" + line + "' is not of type: ");
		}    	
    }
    
    private TimestampedTextAdapter getTimestampedDataValue(String value, int line, Date time) throws PluginException {
		try {
			
			return new TimestampedTextAdapter(value, time);
			
		} catch (NumberFormatException e) {
			this.error(this, this.getPluginId(), this.getId(), "Value '" + value + "' in line '" + line + "' is not of type: ");
			throw new PluginException(PluginExceptionsEnum.InputTypeNotRecognized, this.getPluginId(), "Value '" + value + "' in line '" + line + "' is not of type: ");
		}    	
    }
    
    public Date getTimestampValue(String timestamp, SimpleDateFormat sdf, int line) throws PluginException {
    	try {
			Date ts = sdf.parse(timestamp);
			return ts;
		} catch (ParseException e) {
			this.error(this, this.getPluginId(), this.getId(), "Timestamp format error for value '" + timestamp + "' in line '" + line + " " + e.getMessage());
			throw new PluginException(PluginExceptionsEnum.InputTypeNotRecognized, this.getPluginId(), 
					"Timestamp format error for value '" + timestamp + "' in line '" + line + " " + e.getMessage());

		}
    }


}
